#include <thread>
#include <algorithm>
#include <queue>

#include "node.h"

#include "taskmanager.h"

using namespace std;

static unsigned int barrier_count = 1;

TaskManager::TaskManager(bool shuffle, int jobs) :
    __job_count(jobs),
    __core_jobs(nullptr),
    __shuffle(shuffle)
{
    // Get the maximum CPU core counts.
    int core_count = static_cast<int>(std::thread::hardware_concurrency());
    // For all the invalid jobs, treat as -1.
    if(jobs < 1)
    {
        jobs = -1;
    }
    // Check the -1 settings.
    if(jobs == -1)
    {
        // Automatically use the max core count.
        __job_count = core_count;
    }
    else
    {
        // Check the core information.
        __job_count = jobs > core_count ? core_count : jobs;
    }
    //Set the barrier count.
    barrier_count = static_cast<unsigned int>(__job_count);
}

TaskManager::~TaskManager()
{
    if(__core_jobs)
    {
        // Clear the job mission.
        delete[] __core_jobs;
    }
}

void TaskManager::set_task_list(NodeTaskInfoList task_info_list)
{
    // For single thread, just expand all the task to the first mission.
    if(__job_count == 1)
    {
        // Based on the job count, create the list.
        __core_jobs = new NodeTaskInfoList[1];
        //Append all the data to list.
        __core_jobs[0] = task_info_list;
        // Complete.
        return;
    }
    // Check the size of job count and the list that we have.
    // If we have 12 cores, but only 2 missions, then maximum 2 threads is
    // limited for saving the resources.
    __job_count = qMin(__job_count, task_info_list.size());
    // Based on the job count, create the list.
    __core_jobs = new NodeTaskInfoList[static_cast<size_t>(__job_count)];
    //Update the barrier count.
    barrier_count = static_cast<unsigned int>(__job_count);
    // Based on the job count, allocate the mission to each thread.
    // Sort the task list.
    std::sort(task_info_list.begin(), task_info_list.end());
    // Define the task load structure.
    struct TaskLoad
    {
        int id;
        int load;
        TaskLoad(int task_id) :
            id(task_id),
            load(0)
        {
        }

        bool operator < (const TaskLoad &taskLoad) const
        {
            // In reverse order.
            return this->load > taskLoad.load;
        }
    };
    // Construct the priority queue for all the task loads.
    std::priority_queue<TaskLoad> load_queue;
    for(int i=0; i<__job_count; ++i)
    {
        //Insert the load to queue.
        load_queue.push(TaskLoad(i));
    }
    // Allocate the buckets.
    for(auto task_info : task_info_list)
    {
        // Get the least load task.
        TaskLoad least_load = load_queue.top();
        load_queue.pop();
        // Append the mission to the list.
        __core_jobs[least_load.id].append(task_info);
        // Set the new load for the task.
        least_load.load += task_info.second.size();
        // Put it back to load.
        load_queue.push(least_load);
    }
}

void TaskManager::execute(const NodeTaskInfoList &task_info_list)
{
    // Loop in the provided bucket.
    for(auto task_info : task_info_list)
    {
        //Execute all the task recorded.
        for(auto task : task_info.second)
        {
            // Execute the task.
            (task_info.first->*task)();
        }
    }
}

static unsigned int thread_counter = 0;
static unsigned int barrier_generation = 0;
static std::mutex barrier_mutex;
static std::condition_variable barrier_cv;

void TaskManager::execute_n(const NodeTaskInfoList &task_info_list,
                            int n)
{
    bool use_locker_a = true;
    //Loop and run for multiple times.
    for(int times=0; times<n; ++times)
    {
        // Execute for one time.
        // Loop in the provided bucket.
        TaskManager::execute(task_info_list);
        // Do not need to wait if there is only 1 job.
        if(barrier_count == 1)
        {
            //Ignore all the sync.
            continue;
        }
        // Hanamaru kawaii :)
        // --- Barrier ---
        // Wait for the counter is clear.
        std::unique_lock<std::mutex> barrier_lock(barrier_mutex);
        unsigned gen = barrier_generation;
        if (--thread_counter == 0)
        {
            //Increase generation.
            ++barrier_generation;
            //Reset the counter.
            thread_counter = barrier_count;
            assert(thread_counter != 0);
            //Notify all the threads.
            barrier_cv.notify_all();
        }
        else
        {
            while(gen == barrier_generation)
            {
                barrier_cv.wait(barrier_lock);
            }
        }
    }
}

void TaskManager::run()
{
    // Create thread array.
    std::thread *workers = new std::thread[static_cast<size_t>(__job_count)];
    // Loop and assign mission.
    for(int i=0; i<__job_count; ++i)
    {
        workers[i] = std::thread(&TaskManager::execute, __core_jobs[i]);
    }
    // Wait to join all the tasks.
    for(int i=0; i<__job_count; ++i)
    {
        workers[i].join();
    }
    delete[] workers;
}

void TaskManager::run(int n)
{
    //Initial the counter first,
    thread_counter = barrier_count;
    // Create thread array.
    std::thread *workers = new std::thread[static_cast<size_t>(__job_count)];
    // Loop and assign mission.
    for(int i=0; i<__job_count; ++i)
    {
        workers[i] = std::thread(&TaskManager::execute_n, __core_jobs[i], n);
    }
    //Wait to join all the tasks.
    for(int i=0; i<__job_count; ++i)
    {
        workers[i].join();
    }
    delete[] workers;
}
